package com.shenke.logging.listener;

import com.shenke.logging.model.dto.Event;
import com.shenke.logging.model.entity.OperationLog;
import com.shenke.logging.service.OperationLogService;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 消息消费方
 */
@Component
@Slf4j
@RocketMQMessageListener(
        topic = "operationLog",                     // 1.topic：消息的发送者使用同一个topic
        consumerGroup = "consumer-group",           // 2.group：不用和生产者group相同 ( 在RocketMQ中消费者和发送者组没有关系 )
        selectorExpression = "*",                   // 3.tag：设置为 * 时，表示全部。
        messageModel = MessageModel.CLUSTERING,     // 4.消费模式：默认 CLUSTERING （ CLUSTERING：负载均衡 ）（ BROADCASTING：广播机制 ）
        maxReconsumeTimes = 3                       // 5.最大重试次数：超过后会进入死信队列
)
public class Consumer implements RocketMQListener<Event>, RocketMQPushConsumerLifecycleListener {

    @Resource
    private OperationLogService operationLogService;
    
    // 使用内存缓存实现简单的幂等性检查
    private final ConcurrentHashMap<String, Long> processedMessages = new ConcurrentHashMap<>();

    /**
     * 监听消息
     * @param event 消息事件
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void onMessage(Event event) {
        String messageId = event.getMessageId();
        log.info("收到操作日志消息，消息ID：{}，用户ID：{}，操作：{}", messageId, event.getUserId(), event.getAction());
        
        try {
            // 幂等性检查，避免重复消费
            if (messageId != null && processedMessages.containsKey(messageId)) {
                log.info("消息已处理，跳过：{}", messageId);
                return;
            }
            
            OperationLog operationLog = new OperationLog();
            operationLog.setUserId(event.getUserId());
            operationLog.setAction(event.getAction());
            operationLog.setIp(event.getIp());
            operationLog.setDetail(event.getDetail());
            operationLog.setMessageId(messageId);
            boolean save = operationLogService.save(operationLog);
            if (!save) {
                log.error("日志保存失败，日志信息：{}", operationLog);
                throw new RuntimeException("日志保存失败");
            }
            
            // 记录已处理的消息ID
            if (messageId != null) {
                processedMessages.put(messageId, System.currentTimeMillis());
                // 清理过期缓存数据，避免内存泄漏
                cleanExpiredCache();
            }
            log.info("操作日志保存成功，消息ID：{}", messageId);
        } catch (DuplicateKeyException e) {
            // 处理数据库唯一约束冲突，说明消息已经被处理过
            log.warn("消息重复处理，消息ID：{}", messageId, e);
        } catch (Exception e) {
            // 其他异常记录后抛出，触发重试机制
            log.error("处理操作日志消息失败，消息ID：{}，错误：{}", messageId, e.getMessage(), e);
            throw e;
        }
    }
    
    /**
     * 配置消费者
     * @param consumer 消费者实例
     */
    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        // 配置消费者参数
        consumer.setMaxReconsumeTimes(3);
        consumer.setConsumeTimeout(15000);
        
        // 配置死信队列
        try {
            consumer.subscribe("DLQ_operationLog", "*");
        } catch (Exception e) {
            log.error("订阅死信队列失败", e);
        }
        log.info("RocketMQ消费者配置完成");
    }
    
    /**
     * 清理过期的消息缓存
     */
    private void cleanExpiredCache() {
        long currentTime = System.currentTimeMillis();
        // 清理30分钟前的缓存数据
        final long EXPIRE_TIME = 30 * 60 * 1000;
        processedMessages.entrySet().removeIf(entry -> 
            (currentTime - entry.getValue()) > EXPIRE_TIME);
    }
}
